package xin.alum.aim.coder;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import lombok.SneakyThrows;
import org.springframework.scheduling.annotation.Async;
import xin.alum.aim.model.Transportable;

import java.util.Comparator;
import java.util.concurrent.PriorityBlockingQueue;

abstract class BaseEncoder<T extends Transportable> extends MessageToMessageEncoder<T> {
    protected final InternalLogger logger = InternalLoggerFactory.getInstance(this.getClass());

    private Comparator<T> comparator = (o1, o2) -> o1.getPriority() > o2.getPriority() ? o1.getPriority() : o2.getPriority();

    private PriorityBlockingQueue<T> queue = new PriorityBlockingQueue(512, comparator);

    public BaseEncoder() {

    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if (msg instanceof Transportable) {
            queue.put((T) msg);
            send(ctx, promise);
        } else if (ctx.channel().isWritable()) {
            super.write(ctx, msg, promise);
            super.flush(ctx);
        }
    }

    @SneakyThrows
    @Async
    public void send(ChannelHandlerContext ctx, ChannelPromise promise) {
        while (ctx.channel().isWritable() && queue.iterator().hasNext()) {
            T m = queue.poll();
            if (m == null)
                break;
            super.write(ctx, m, promise);
        }
        super.flush(ctx);
    }

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        queue.clear();
        super.close(ctx, promise);
    }
}
